package com.shujia.hbase;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.TableName;
import org.apache.hadoop.hbase.client.*;
import org.apache.hadoop.hbase.filter.*;
import org.apache.hadoop.hbase.util.Bytes;
import org.junit.Before;
import org.junit.Test;

import java.io.BufferedReader;
import java.io.FileReader;
import java.io.IOException;

public class Demo4Fliter {
    private Connection connection;

    @Before
    public void startup() {
        Configuration configuration = new Configuration();

        //指定zk的连接地址
        configuration.set("hbase.zookeeper.quorum", "master:2181,node1:2181,node2:2181");

        //创建连接
        try {
            connection = ConnectionFactory.createConnection(configuration);
        } catch (IOException e) {
            e.printStackTrace();
        }

    }

    @Test
    public void columnValueFilter() throws IOException {
        Table student = connection.getTable(TableName.valueOf("student"));


        Scan scan = new Scan();

        //正则比较器
        RegexStringComparator regexStringComparator = new RegexStringComparator("文科*");

        //字符串包含比较器
        SubstringComparator substringComparator = new SubstringComparator("理科");

        //二进制前缀比较器
        BinaryPrefixComparator binaryPrefixComparator = new BinaryPrefixComparator("文科".getBytes());


        //单值比较
        BinaryComparator binaryComparator = new BinaryComparator("文科一班".getBytes());


        //列值过滤器
        SingleColumnValueFilter singleColumnValueFilter = new SingleColumnValueFilter(
                "info".getBytes(),//列簇
                "clazz".getBytes(),//列名
                CompareFilter.CompareOp.EQUAL,// 比较方式
                binaryComparator//比较器
        );


        //在全表扫描的基础上增加过滤器，过滤器运行在regionserver中，可以减少数据传入的数据量
        scan.setFilter(singleColumnValueFilter);


        ResultScanner scanner = student.getScanner(scan);

        Result res;

        while ((res = scanner.next()) != null) {
            String id = Bytes.toString(res.getRow());
            String name = Bytes.toString(res.getValue("info".getBytes(), "name".getBytes()));
            String age = Bytes.toString(res.getValue("info".getBytes(), "age".getBytes()));
            String gender = Bytes.toString(res.getValue("info".getBytes(), "gender".getBytes()));
            String clazz = Bytes.toString(res.getValue("info".getBytes(), "clazz".getBytes()));
            System.out.println(id + "\t" + name + "\t" + age + "\t" + gender + "\t" + clazz);
        }


    }


    @Test
    public void RowFilter() throws IOException {
        Table student = connection.getTable(TableName.valueOf("student"));


        Scan scan = new Scan();


        //二进制前缀比较器
        BinaryPrefixComparator binaryPrefixComparator = new BinaryPrefixComparator("15001009".getBytes());


        //行键过滤器
        //效率最高，有序的
        RowFilter rowFilter = new RowFilter(CompareFilter.CompareOp.EQUAL, binaryPrefixComparator);


        //增加过滤器
        scan.setFilter(rowFilter);


        ResultScanner scanner = student.getScanner(scan);

        Result res;

        while ((res = scanner.next()) != null) {
            String id = Bytes.toString(res.getRow());
            String name = Bytes.toString(res.getValue("info".getBytes(), "name".getBytes()));
            String age = Bytes.toString(res.getValue("info".getBytes(), "age".getBytes()));
            String gender = Bytes.toString(res.getValue("info".getBytes(), "gender".getBytes()));
            String clazz = Bytes.toString(res.getValue("info".getBytes(), "clazz".getBytes()));
            System.out.println(id + "\t" + name + "\t" + age + "\t" + gender + "\t" + clazz);
        }

    }


    @Test
    public void putAll() throws Exception {
        //获取表的示例对象
        Table student = connection.getTable(TableName.valueOf("student"));


        //读取学生表数据
        BufferedReader bufferedReader = new BufferedReader(new FileReader("data/students.txt"));

        String line;
        while ((line = bufferedReader.readLine()) != null) {
            String[] split = line.split(",");
            String id = split[0];
            String name = split[1];
            String age = split[2];
            String gender = split[3];
            String clazz = split[4];


            //将性别作为rowkey的前缀，提高查询的效率
            /**
             * 二级索引： 建立行键于列值的映射关系
             *
             */
            String rowkey = gender + "_" + id;
            //构建put 以学号作为rowkey
            Put put = new Put(rowkey.getBytes());


            //增加列
            put.addColumn("info".getBytes(), "name".getBytes(), name.getBytes());
            put.addColumn("info".getBytes(), "age".getBytes(), age.getBytes());
            put.addColumn("info".getBytes(), "gender".getBytes(), gender.getBytes());
            put.addColumn("info".getBytes(), "clazz".getBytes(), clazz.getBytes());

            //插入一条数据
            student.put(put);

        }

    }


    @Test
    public void genderFIlter() throws IOException {
        Table student = connection.getTable(TableName.valueOf("student"));


        Scan scan = new Scan();


        //二进制前缀比较器
        BinaryPrefixComparator binaryPrefixComparator = new BinaryPrefixComparator("男".getBytes());


        //行键过滤器
        //效率最高，有序的
        RowFilter rowFilter = new RowFilter(CompareFilter.CompareOp.EQUAL, binaryPrefixComparator);


        //增加过滤器
        scan.setFilter(rowFilter);


        ResultScanner scanner = student.getScanner(scan);

        Result res;

        while ((res = scanner.next()) != null) {
            String id = Bytes.toString(res.getRow());
            String name = Bytes.toString(res.getValue("info".getBytes(), "name".getBytes()));
            String age = Bytes.toString(res.getValue("info".getBytes(), "age".getBytes()));
            String gender = Bytes.toString(res.getValue("info".getBytes(), "gender".getBytes()));
            String clazz = Bytes.toString(res.getValue("info".getBytes(), "clazz".getBytes()));
            System.out.println(id + "\t" + name + "\t" + age + "\t" + gender + "\t" + clazz);
        }

    }


}
